package com.redis.study.stream.run;

import com.redis.study.stream.consumers.EmailStreamMessageListener;
import com.redis.study.stream.consumers.StationStreamMessageListener;
import com.redis.study.stream.util.IdUtil;
import com.redis.study.stream.util.SnowflakeIdWorker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamMessageListenerContainerOptions;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Vector;

@Slf4j
@Component
public class StreamConsumerRunner implements ApplicationRunner, DisposableBean {

    @Autowired
    RedisConnectionFactory redisConnectionFactory;

    @Autowired
    ApplicationContext applicationContext;

    @Autowired
    ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    private Vector<StreamMessageListenerContainer<String, MapRecord<String, String, String>>> listenerContainers = new Vector<>();
    @Override
    public void run(ApplicationArguments args) {
        try {
            stringRedisTemplate.opsForStream().createGroup("abc", "station-group");
            stringRedisTemplate.opsForStream().createGroup("abc", "email-group");
        } catch (Exception e) {
            log.info("组存在");
        }

        Group blogGroup = new Group();
        blogGroup.setGroupName("station-group");
        blogGroup.setConsumers(Arrays.asList("xf1","xf2"));
        Group emailGroup = new Group();
        emailGroup.setGroupName("email-group");
        emailGroup.setConsumers(Arrays.asList("xf1","xf2"));

        StreamEntity streamEntity = new StreamEntity();
        streamEntity.setKeyName("abc");
        streamEntity.setGroups(Arrays.asList(blogGroup,emailGroup));

        for (Group group : streamEntity.getGroups()) {
            for (String consumer : group.getConsumers()) {
                // 根据配置对象创建监听容器对象
                StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer = StreamMessageListenerContainer
                        .create(this.redisConnectionFactory, options(group.getGroupName(),consumer));
                Consumer consumerOne = Consumer.from(group.getGroupName(), consumer);
                StreamOffset<String> stringStreamOffset = StreamOffset.create(streamEntity.getKeyName(), ReadOffset.lastConsumed());
                streamMessageListenerContainer.receive(consumerOne,stringStreamOffset, getType(group.getGroupName()));
                streamMessageListenerContainer.start();
                listenerContainers.add(streamMessageListenerContainer);
            }
        }
        log.info("有{}监听已启动",listenerContainers.size());
        // 模拟生产数据
        Map<String,String> map = new LinkedHashMap<>();
        map.put(String.valueOf(IdUtil.genId()), String.valueOf(IdUtil.genId()));
        stringRedisTemplate.opsForStream().add("abc", map);
    }

    /**
     *  通过对应的组名来获取对应消息处理对象
     * @param type 组名
     * @return 消息处理对象
     */
    public StreamListener<String, MapRecord<String, String, String>> getType(String type){
        if ("station-group".equals(type)){
            return applicationContext.getBean(StationStreamMessageListener.class);
        }
        if ("email-group".equals(type)){
            return applicationContext.getBean(EmailStreamMessageListener.class);
        }
        return null;
    }

    /**
     *  创建StreamMessageListenerContainerOptions对象
     */
    private StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options(String group,String consumer){
        // 设置线程的前缀
        threadPoolTaskExecutor.setThreadNamePrefix(group+"-");
        // 设置线程执行的线程名
        threadPoolTaskExecutor.setThreadGroupName(consumer);
        return  StreamMessageListenerContainerOptions
                .builder()
                // 一次性最多拉取多少条消息
                .batchSize(1)
                // 执行消息轮询的执行器
                .executor(this.threadPoolTaskExecutor)
                // 超时时间，设置为0，表示不超时（超时后会抛出异常）
                .pollTimeout(Duration.ZERO)
                // 序列化器
                .serializer(new StringRedisSerializer())
                .build();
    }


    @Override
    public void destroy(){
        log.info("程序：destroy、关闭{}个监听器",listenerContainers.size());
        this.listenerContainers.forEach(StreamMessageListenerContainer::stop);
    }
}
